package com.cbh.amadeus.client.factory;

import com.cbh.amadeus.client.config.SubscriberConfig;
import com.cbh.amadeus.client.domain.MessageRepositoryModel;
import com.cbh.amadeus.client.receiver.tool.MessageRepositoryModelTool;
import com.cbh.amadeus.client.utils.ChannelReference;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * @author ChenBiHui
 * @date 2025/5/9
 */
@Slf4j
public class MessageResendFactory implements InitializingBean {

    private static final Deque<Thread> REACH = new ArrayDeque<>(1);
    private final SubscriberConfig config;
    private final MessageRepositoryModelTool tool;


    public MessageResendFactory(SubscriberConfig config, MessageRepositoryModelTool tool) {
        this.config = config;
        this.tool = tool;
    }

    public static void call() {
        Thread reach = REACH.peekFirst();
        if (reach != null && reach.isAlive()) {
            LockSupport.unpark(reach);
        }
    }

    @Override
    public void afterPropertiesSet() {
        Thread resend = new Thread(() -> {
            while (true) {
                Long resendTimeSeconds = config.getMessageResendTimeSeconds();
                try {
                    TimeUnit.SECONDS.sleep(resendTimeSeconds);
                } catch (InterruptedException ignored) {
                }
                List<MessageRepositoryModel> messages = tool.list(config.getMessageResendLimit(), MessageRepositoryModel.MessageRepositoryEnum.ACTIVE);
                if (messages == null || messages.isEmpty()) {
                    log.info("===== Resend factory can't find any active messages, has park =====");
                    LockSupport.park();
                    continue;
                }
                long now = System.currentTimeMillis();
                for (MessageRepositoryModel message : messages) {
                    if (config.getMessageAbandonTimes() <= message.getReject()) {
                        MessageAbandonFactory.put(message);
                        continue;
                    }
                    if (now < message.getAchieve()) {
                        continue;
                    }
                    Optional.ofNullable(ChannelReference.get())
                            .ifPresent(channel -> channel.writeAndFlush(resend(message)));
                }
            }
        });
        resend.setName("amadeus-message-resend");
        resend.setDaemon(true);
        resend.start();
        REACH.push(resend);
        log.info("===== AmadeusMessageResend started =====");
    }

    private AmadeusMessageWrap resend(MessageRepositoryModel message) {
        AmadeusMessageWrap wrap = new AmadeusMessageWrap();
        AmadeusMessage amadeus = new AmadeusMessage();
        wrap.setMagic(AmadeusMessageProtocol.MAGIC);
        wrap.setType(MessageType.RESEND.getType().byteValue());
        wrap.setApproach(message.getApproach());
        wrap.setAchieve(message.getAchieve());
        wrap.setMessage(amadeus);
        amadeus.setProvider(message.getProvider());
        amadeus.setTopic(message.getTopic());
        amadeus.setMessageId(message.getMessageId());
        amadeus.setPayload(message.getPayload());
        return wrap;
    }
}
